package cn.lazar.homework


import org.apache.spark.rdd.{CoGroupedRDD, RDD}
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext, TaskContext}


object JoinDemo {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")

    val sc = new SparkContext(conf)

    sc.setLogLevel("WARN")

    val random = scala.util.Random

    val col1 = Range(1, 50).map(idx => (random.nextInt(10), s"user$idx"))

    val col2 = Array((0, "BJ"), (1, "SH"), (2, "GZ"), (3, "SZ"), (4, "TJ"), (5, "CQ"), (6, "HZ"), (7, "NJ"), (8, "WH"), (0,"CD"))


    val rdd1: RDD[(Int, String)] = sc.makeRDD(col1)
    println(rdd1.getNumPartitions)
    println(rdd1.partitioner)

    val rdd2: RDD[(Int, String)] = sc.makeRDD(col2)
    println(rdd2.partitioner)

    //直接cogroup
//    val cg = new CoGroupedRDD[Int](Seq(rdd1, rdd2), new HashPartitioner(12))
//    println(cg.dependencies)
   // List(org.apache.spark.ShuffleDependency@5e840abf, org.apache.spark.ShuffleDependency@56de6d6b)

//    val partitioner = new HashPartitioner(3)
//    val cg = new CoGroupedRDD[Int](Seq(rdd1.partitionBy(partitioner), rdd2.partitionBy(new HashPartitioner(3))), partitioner)
//    println(cg.dependencies)
//    List(org.apache.spark.OneToOneDependency@7a18e8d, org.apache.spark.ShuffleDependency@3028e50e)
//
    val rdd3: RDD[(Int, (String, String))] = rdd1.join(rdd2)

    println(rdd3.count())
    println(rdd3.dependencies)
    println(rdd3.getNumPartitions)
    println(rdd3.partitioner)

    val rdd4: RDD[(Int, (String, String))] = rdd1.partitionBy(new HashPartitioner(3)).join(rdd2.partitionBy(new HashPartitioner(3)))


    println(rdd4.count())
    println(rdd4.dependencies)
    println(rdd4.getNumPartitions)

    Thread.sleep(1000000)
    sc.stop()

  }

}

// 输出结果
// List(org.apache.spark.OneToOneDependency@21325036) 宽依赖
// List(org.apache.spark.OneToOneDependency@267517e4) 窄依赖
// 因为join操作产生了三个RDD cogroupRDD MAPpartionerRDD MAPpartionerRDD ,最后返回的PairRDD 是 OneToOneDependency 窄依赖，但是
// 中的关键步骤 cogroupRDD 有可能是ShuffleDependency 宽依赖，当join前的RDD未分区的时候
// 结论
// join 之前的RDD未分区则为宽依赖，已经分区的为窄依赖


